package mq

import (
	"MessageSendService/app/common/log"
	"MessageSendService/app/common/models"
	"MessageSendService/app/job/etc"
	"MessageSendService/app/job/internal/task"
	"context"
	"encoding/json"
	"fmt"
	"go.uber.org/zap"
	"time"

	"github.com/segmentio/kafka-go"
)

var (
	topic = "sujinke"
)

func Init() {

	//初始化消费者
	consumer := kafka.NewReader(kafka.ReaderConfig{
		Brokers:                []string{etc.Config.KafkaUrl}, // broker地址 数组
		GroupID:                "test",                        // 消费者组id，每个消费者组可以消费kafka的完整数据，但是同一个消费者组中的消费者根据设置的分区消费策略共同消费kafka中的数据
		GroupTopics:            nil,
		Topic:                  topic, // 消费哪个topic
		Partition:              0,
		Dialer:                 nil,
		QueueCapacity:          0,
		MinBytes:               0,
		MaxBytes:               0,
		MaxWait:                0,
		ReadBatchTimeout:       0,
		ReadLagInterval:        0,
		GroupBalancers:         nil,
		HeartbeatInterval:      0,
		CommitInterval:         time.Second, // offset 上报间隔
		PartitionWatchInterval: 0,
		WatchPartitionChanges:  false,
		SessionTimeout:         0,
		RebalanceTimeout:       0,
		JoinGroupBackoff:       0,
		RetentionTime:          0,
		StartOffset:            kafka.FirstOffset, // 仅对新创建的消费者组生效，从头开始消费，工作中可能更常用从最新的开始消费kafka.LastOffset
		ReadBackoffMin:         0,
		ReadBackoffMax:         0,
		Logger:                 nil,
		ErrorLogger:            nil,
		IsolationLevel:         0,
		MaxAttempts:            0,
		OffsetOutOfRangeError:  false,
	})

	// 开始消费消息
	ReadMessage(context.Background(), consumer)

}

// 消费消息
func ReadMessage(ctx context.Context, consumer *kafka.Reader) {
	// 消费者应该通过协程一直开着，一直消费
	for {
		if msg, err := consumer.ReadMessage(ctx); err != nil {
			fmt.Println(fmt.Sprintf("读kafka失败，err:%v", err))
			continue // 当前消息读取失败时，并不退出for终止所有后续消费，而是跳过该消息即可
		} else {
			taskInfo := &models.TaskInfo{}
			err := json.Unmarshal(msg.Value, taskInfo)
			if err != nil {
				fmt.Println(fmt.Sprintf("json unmarshal msg value err，msg:%v,err:%v", taskInfo, err))
				continue // 当前消息处理失败时，并不退出for终止所有后续消费，而是跳过该消息即可
			}
			curTask := task.NewTask(*taskInfo)
			err = curTask.Run()
			if err != nil {
				log.GetLogger().Info("消息发送失败", zap.Any("taskInfo", taskInfo), zap.Any("err", err))
			} else {
				log.GetLogger().Info("消息发送成功", zap.Any("taskInfo", taskInfo), zap.Any("err", err))
			}
		}
	}
}
